package com.liuqi.knet.rpc.cluster;

import com.liuqi.knet.remoting.ChannelHandler;
import com.liuqi.knet.remoting.Client;
import com.liuqi.knet.remoting.RemotingException;
import com.liuqi.knet.remoting.exchange.ResponseFuture;
import com.liuqi.knet.remoting.transport.AbstractClient;
import com.liuqi.loggers.api.Logger;
import com.liuqi.loggers.facade.LoggersFacade;
import java.lang.reflect.Constructor;
import java.net.InetSocketAddress;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author liuqi
 * @date 2025/3/6 15:27
 **/
public abstract class AbstractCluster implements Cluster {

    private final static Logger logger = LoggersFacade.getLogger(AbstractCluster.class);

    private final LoadBalance loadBalance;

    private final ChannelHandler handler;

    private final Lock connectLock = new ReentrantLock();

    private final int sendRetry;

    private final ClusterConfig clusterConfig;

    public AbstractCluster(ClusterConfig clusterConfig) {
        this.clusterConfig = clusterConfig;
        this.loadBalance = clusterConfig.getLoadBalance();
        this.handler = clusterConfig.getChannelHandler();
        if (loadBalance == null) {
            throw new NullPointerException("loadBalance is null. please use [ClusterConfig.INSTANCE.setLoadBalance(LoadBalance)]为其赋值");
        }
        if (handler == null) {
            throw new NullPointerException("handler is null. please use [ClusterConfig.INSTANCE.setChannelHandler(ChannelHandler)]为其赋值");
        }
        this.sendRetry = clusterConfig.getSendRetry();
    }

    public LoadBalance getLoadBalance() {
        return loadBalance;
    }

    @Override
    public ResponseFuture request(Object request, int timeout) throws RemotingException {
        int retry = sendRetry + 1;
        try {
            for (int i = 0; i < retry; i++) {
                try {
                    Client client = getClient();
                    return client.request(request, timeout);
                } catch (RemotingException e) {
                    if (i == retry - 1) {
                        logger.error("消息发送失败, 已到达最大重试次数: " + clusterConfig.getSendRetry() + ", ErrorMsg: " + e.getMessage(), e);
                        throw e;
                    }
                    logger.error("消息发送失败, 进行重试: 第[" + (i + 1) + "]次重试, ErrorMsg: " + e.getMessage(), e);
                }
            }
        } catch (RemotingException e) {
            throw e;
        }
        return null;
    }

    @Override
    public void send(Object message, boolean sent) throws RemotingException {
        int retry = sendRetry + 1;
        try {
            for (int i = 0; i < retry; i++) {
                try {
                    Client client = getClient();
                    client.send(message, sent);
                } catch (RemotingException e) {
                    if (i == retry - 1) {
                        logger.error("消息发送失败, 已到达最大重试次数: " + clusterConfig.getSendRetry() + ", ErrorMsg: " + e.getMessage(), e);
                        throw e;
                    }
                    logger.error("消息发送失败, 进行重试: 重试次数: " + (i + 1) + ", ErrorMsg: " + e.getMessage(), e);
                }
            }
        } catch (RemotingException e) {
            throw e;
        }
    }

    /**
     * 创建客户端连接
     *
     * @param node
     * @return
     * @throws RemotingException
     */
    protected Client createClient(LoadBalanceNode.Node node) throws RemotingException {
        connectLock.lock();
        AbstractClient client = null;
        try {
            InetSocketAddress remote = node.getRemote();
            Constructor<? extends AbstractClient> constructor =
                    clusterConfig.getClientImpl()
                            .getConstructor(InetSocketAddress.class, ChannelHandler.class);
            client = constructor.newInstance(remote, handler);
            if (client.isConnected()) {
                return client;
            }
            throw new RemotingException("客户端建立连接失败");
        } catch (Throwable t) {
            if (client != null) {
                try {
                    client.close();
                } catch (Throwable tt) {
                    logger.error(tt.getMessage());
                }
            }
            if (t instanceof RemotingException) {
                throw (RemotingException) t;
            } else {
                Throwable cause = t.getCause();
                if (cause instanceof RemotingException) {
                    throw (RemotingException) cause;
                } else {
                    throw new RemotingException(t.getMessage(), t);
                }
            }
        } finally {
            connectLock.unlock();
        }
    }

    /**
     * @return  ChannelHandler
     */
    protected ChannelHandler getHandler() {
        return this.handler;
    }

    protected abstract Client getClient() throws RemotingException;


}
